package org.tao.lightningmq.nameserver.handler;

import com.alibaba.fastjson2.JSON;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.internal.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.tao.lightningmq.common.dto.HeartBeatDTO;
import org.tao.lightningmq.common.dto.PullBrokerIpDTO;
import org.tao.lightningmq.common.dto.ServiceRegistryReqDTO;
import org.tao.lightningmq.common.enums.NameServerEventCode;
import org.tao.lightningmq.common.event.EventBus;
import org.tao.lightningmq.common.event.model.BaseEvent;
import org.tao.lightningmq.common.tcp.TcpMsg;
import org.tao.lightningmq.nameserver.event.model.HeartBeatEvent;
import org.tao.lightningmq.nameserver.event.model.PullBrokerIpEvent;
import org.tao.lightningmq.nameserver.event.model.RegistryEvent;
import org.tao.lightningmq.nameserver.event.model.UnRegistryEvent;

import java.net.InetSocketAddress;

/**
 * @Author lht
 * @date 2025/1/7 - 19:34
 * @description:
 */
@ChannelHandler.Sharable
public class TcpNettyServerHandler extends SimpleChannelInboundHandler<TcpMsg> {

    private static final Logger LOGGER = LoggerFactory.getLogger(TcpNettyServerHandler.class);

    private final EventBus eventBus;

    public TcpNettyServerHandler(EventBus eventBus) {
        LOGGER.info("TcpNettyServerHandler初始化");
        this.eventBus = eventBus;
        eventBus.init();
    }

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TcpMsg tcpMsg) throws Exception {
        int code = tcpMsg.getCode();
        byte[] body = tcpMsg.getBody();
        BaseEvent event = null;
        if (NameServerEventCode.REGISTRY.getCode() == tcpMsg.getCode()) {
            ServiceRegistryReqDTO serviceRegistryReqDTO = JSON.parseObject(body, ServiceRegistryReqDTO.class);
            RegistryEvent registryEvent = new RegistryEvent(serviceRegistryReqDTO);
            if (StringUtil.isNullOrEmpty(serviceRegistryReqDTO.getIp())) {
                InetSocketAddress inetSocketAddress = (InetSocketAddress) ctx.channel().remoteAddress();
                registryEvent.setPort(inetSocketAddress.getPort());
                registryEvent.setIp(inetSocketAddress.getHostString());
            } else {
                registryEvent.setPort(serviceRegistryReqDTO.getPort());
                registryEvent.setIp(serviceRegistryReqDTO.getIp());
            }
            event = registryEvent;
        } else if (NameServerEventCode.HEART_BEAT.getCode() == tcpMsg.getCode()) {
            HeartBeatDTO heartBeatDTO = JSON.parseObject(body, HeartBeatDTO.class);
            HeartBeatEvent heartBeatEvent = new HeartBeatEvent();
            heartBeatEvent.setMsgId(heartBeatDTO.getMsgId());
            event = heartBeatEvent;
        } else if (NameServerEventCode.PULL_BROKER_IP_LIST.getCode() == tcpMsg.getCode()) {
            PullBrokerIpEvent pullBrokerIpEvent = new PullBrokerIpEvent();
            PullBrokerIpDTO pullBrokerIpDTO = JSON.parseObject(body, PullBrokerIpDTO.class);
            pullBrokerIpEvent.setRole(pullBrokerIpDTO.getRole());
            pullBrokerIpEvent.setBrokerClusterGroup(pullBrokerIpDTO.getBrokerClusterGroup());
            pullBrokerIpEvent.setMsgId(pullBrokerIpDTO.getMsgId());
            event = pullBrokerIpEvent;
        } else {
            ctx.close();
            throw new IllegalArgumentException("Unknown code: " + code);
        }
        if (event != null) {
            event.setContext(ctx);
            eventBus.publish(event);
        }
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        UnRegistryEvent unRegistryEvent = new UnRegistryEvent();
        unRegistryEvent.setContext(ctx);
        eventBus.publish(unRegistryEvent);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        LOGGER.warn("Netty exception caught: ", cause);
    }
}
